package com.cuit.remoting.transport.netty.client;

import com.cuit.factory.SingleFactory;
import com.cuit.retryies.RetryMap;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;

import com.cuit.retryies.RetryPolicy;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class ConnectionWatchHandler extends ChannelInboundHandlerAdapter implements TimerTask {

    private volatile boolean reconnect = true;
    private Timer timer;
    private final ChannelProvider channelProvider;
    private final Bootstrap bootstrap;
    private RetryPolicy retryPolicy;
    private RetryMap retriesMap;
    private InetSocketAddress inetSocketAddress;

    public ConnectionWatchHandler(Bootstrap bootstrap, HashedWheelTimer timer, RetryPolicy retryPolicy) {
        this.bootstrap = bootstrap;
        this.channelProvider = SingleFactory.getSingle(ChannelProvider.class);
        this.retriesMap = SingleFactory.getSingle(RetryMap.class);
        this.timer = timer;
        this.retryPolicy = retryPolicy;
    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        AtomicInteger atomicInteger = retriesMap.get(inetSocketAddress.toString());
        atomicInteger.set(0);
        log.info("当前[{}]链路已经激活，重新尝试链接次数重新置为0", inetSocketAddress.toString());
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        reconnect();

    }
    public void reconnect(){
        log.info(this.toString());
        AtomicInteger retries = retriesMap.get(inetSocketAddress.toString());
        if (retryPolicy.allowRetry(retries.get())) {
            long sleepTime = retryPolicy.getSleepTime(retries.get());
            log.error("链路[{}]已关闭，重新尝试连接,第{}次连接,等待时间{}",inetSocketAddress.toString(), retries.incrementAndGet(), sleepTime);
            timer.newTimeout(this, sleepTime, TimeUnit.MILLISECONDS);
        }
    }

    @Override
    public void run(Timeout timeout) throws Exception {
        ChannelFuture channelFuture;
        synchronized (bootstrap){
            channelFuture = bootstrap.connect(inetSocketAddress);
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    boolean success = future.isSuccess();
                    if (!success){
                        log.error("[{}]重连失败", inetSocketAddress.toString());
                        reconnect();
                    } else{
                        log.info("[{}]重连成功", inetSocketAddress.toString());
                        channelProvider.set(inetSocketAddress, future.channel());
                    }
                }
            });
        }
    }
}
